package com.mydemo.springlearning.functions.threadpool;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.time.LocalTime;
import java.util.concurrent.*;

/**
 * @author :
 * @date : 2021/8/4 15:55
 * @description: 阿里巴巴推荐，不自己启动线程。而使用线程池处理
 * 线程池处理类
 */
public class ThreadPoolService {

    /**
     * 自定义线程名称,方便的出错的时候溯源
     */
    private static final ThreadFactory NAMED_THREAD_FACTORY = new ThreadFactoryBuilder().setNameFormat("CALLBACK_POOL-%d").build();
    /**
     * 获取当前运行机器的可用CPU核心数
     */
    private final static int POLLER_THREAD_COUNT = Runtime.getRuntime().availableProcessors();

    /**
     * corePoolSize    线程池中即使空闲也不会杀死的线程数目，除非 {@code allowCoreThreadTimeOut} 被配置<br>
     * maximumPoolSize 线程池中允许的最大线程数量<br>
     * keepAliveTime   当线程数大于核心时，多出的空闲线程终止前等待新任务的最长时间<br>
     * unit            {@code keepAliveTime} 的时间单位<br>
     * workQueue       用来储存等待执行任务的队列<br>
     * 阻塞队列：<br>
     * 　　ArrayBlockingQueue ：一个由数组结构组成的有界阻塞队列。<br>
     * 　　LinkedBlockingQueue ：一个由链表结构组成的有界阻塞队列。 <br>
     * 　　PriorityBlockingQueue ：一个支持优先级排序的无界阻塞队列。<br>
     * 　　DelayQueue： 一个使用优先级队列实现的无界阻塞队列。<br>
     * 　　SynchronousQueue： 一个不存储元素的阻塞队列。<br>
     * 　　LinkedTransferQueue： 一个由链表结构组成的无界阻塞队列。<br>
     * 　　LinkedBlockingDeque： 一个由链表结构组成的双向阻塞队列。<br>
     * <br>
     * <p>
     * threadFactory   创建线程的工厂类<br>
     * handler         拒绝策略类,当线程池数量达到上线并且workQueue队列长度达到上限时就需要对到来的任务做拒绝处理<br>
     * 执行流程：<br>
     * 0. 线程池创建，准备核心线程，准备接受任务
     * 1. 有请求时，创建线程执行任务，<br>
     * 2. 当线程数量大于corePoolSize时，请求加入阻塞队列里，空闲core会去阻塞队列获取任务执行<br>
     * 3. 当队列满了(workQueue.size())时，接着创建线程，直到线程数等于maximumPoolSize。<br>
     * 4. 当线程数满了，任务仍增加的时候，线程池开始执行拒绝策略。<br>
     * 换言之，线程池最多同时并行执行maximumPoolSize的线程，最多接受corePoolSize+maximumPoolSize+workQueue.size()的任务。多余的默认采用AbortPolicy会丢弃。
     * <br>
     * 拒绝策略：<br>
     * 　　ThreadPoolExecutor.AbortPolicy: 丢弃任务并抛出RejectedExecutionException异常。 (默认)<br>
     * 　　ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。<br>
     * 　　ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务。（重复此过程）<br>
     * 　　ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务。<br>
     * 案例:<br>
     * 一个线程池，core 7，max 20, queue 50, 并发100 <br>
     * 7个立即执行，50个进入队列，再开13个线程，剩下30个执行拒绝策略
     * 如果不想抛弃，执行{@link java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy}会在当前线程直接执行任务
     */
    private static final ExecutorService service = new ThreadPoolExecutor(
            POLLER_THREAD_COUNT,
            POLLER_THREAD_COUNT * 8,
            10L,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(64),
            NAMED_THREAD_FACTORY,
            new ThreadPoolExecutor.CallerRunsPolicy()
    );

    /**
     * 获取线程池
     *
     * @return 线程池
     */
    public static ExecutorService getEs() {
        return service;
    }

    /**
     * 使用线程池创建线程并异步执行任务
     *
     * @param r 任务
     */
    public static void newTask(Runnable r) {
        service.execute(r);
    }

    public static void shutdown() {
        service.shutdown();
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        function3();
        multiRunCombine();
    }

    private static void func0() {
        // 1.
     /*   for (int i = 0;i<20;i++){
            int finalI = i;
            Runnable r = new Runnable() {
                @Override
                public void run() {
                    System.out.println("当前线程为:"+ finalI+":"+Thread.currentThread().getName());
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            newTask(r);
        }*/
        // 2.
        for (int i = 0; i < 20; i++) {
            int finalI = i;
            service.execute(() -> {
                System.out.println("当前任务为:" + finalI + ", 执行线程是：" + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
//        service.shutdown();
    }


    public static void function1() {
        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
            System.out.println("thread name：" + Thread.currentThread().getName() + ",(id)-" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
        }, service);
        System.out.println("func ends");
    }

    public static void function2() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("thread name：" + Thread.currentThread().getName() + ",(id)-" + Thread.currentThread().getId());
            int i = 10 / 0;
            System.out.println("运行结果：" + i);
            return i;
        }, service).whenComplete((result, exception) -> {
            //虽然得到异常信息但没法得到返回参数
            System.out.println("task complete-result:" + result);
            System.out.println("task complete-exception:" + exception);
        }).exceptionally((throwable -> {
            //异常返回默认值
            return -1;
        })).handle((res, throwable) -> {
            if (res != null) {
                return res * 2;
            }
            if (throwable != null) {
                return -2;
            }
            return 0;
        });
        System.out.println("func ends :" + completableFuture.get());
    }

    /**
     * 线程串行化
     * 1)、thenRun 不能获取到上一步执行结果，无返回值
     *      .thenRunAsync(() -> {
     *             System.out.println("任务2启动了");
     *       }, service);
     * 2)、thenAccept 能接收上一步返回值，但是无返回值
     * 3)、thenApply 既能接收上一步返回值，也能有返回值
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void function3() throws ExecutionException, InterruptedException {
//        CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(() -> {
//            System.out.println("thread name：" + Thread.currentThread().getName() + ",(id)-" + Thread.currentThread().getId());
//            int i = 10 / 5;
//            System.out.println("运行结果：" + i);
//            return i;
//        }, service).thenAcceptAsync((res) -> {
//            System.out.println("任务2启动了..."+res);
//        }, service);
//        System.out.println("func ends :" + completableFuture.get());
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("thread name：" + Thread.currentThread().getName() + ",(id)-" + Thread.currentThread().getId());
            int i = 10 / 5;
            System.out.println("运行结果：" + i);
            return i;
        }, service).thenApplyAsync((res ) -> {
            System.out.println("任务2启动了..."+res);
            return "res :"+res;
        }, service);
        System.out.println("func ends :" + completableFuture.get());
    }
    public static void multiRunCombine() throws ExecutionException, InterruptedException {
//        CompletableFuture<Void> fu = CompletableFuture.supplyAsync(() -> {
//            System.out.println("thread name：" + Thread.currentThread().getName() + ",(id)-" + Thread.currentThread().getId());
//            int i = 10 / 5;
//            System.out.println("运行结果：" + i);
//            return i;
//        }, service).thenAcceptAsync((res) -> {
//            System.out.println("任务2启动了..."+res);
//        }, service);
//        System.out.println("func ends :" + fu.get());


        CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
            System.out.println("thread name：" + Thread.currentThread().getName() + ",(id)-" + Thread.currentThread().getId());
            int i = 2222 / 2;
            System.out.println("运行结果：" + i);
            return i;
        }, service);
        CompletableFuture<Integer> future02 = CompletableFuture.supplyAsync(() -> {
            System.out.println("thread name：" + Thread.currentThread().getName() + ",(id)-" + Thread.currentThread().getId());
            int i = 4444 / 2;
            System.out.println("运行结果：" + i);
            return i;
        }, service);
//       // 1. run 不能感知两者的运行结果，只能保证运行顺序
//        future01.runAfterBothAsync(future02, () -> {
//            System.out.println("任务3开始。。。");
//        });

//        // 2. accept对两者结果可以感知，并保证顺序,没有返回值
//        future01.thenAcceptBothAsync(future02, (future01Result, future02Result) -> {
//            System.out.println("任务3开始...之前的结果:"+"f1-"+future01Result+",f2-"+future02Result);
//        }, service);

        //3. combine对两者结果可以感知，并保证顺序,有返回值
//        CompletableFuture<String> combineAsync = future01.thenCombineAsync(future02, (f1, f2) -> {
//            return f1 + "," + f2;
//        }, service);
//        System.out.println("func ends :" +combineAsync.get() );

        // 4. run+Either run 不能感知两者的运行结果，只要一个完成即执行接下来
        future01.runAfterEitherAsync(future02,()->{
            System.out.println("二者之一已经完成");
        } , service);
    }
    public static void test1( ) throws ExecutionException, InterruptedException {
        CompletableFuture<String> futureImg = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("查询商品的图片信息"+ LocalTime.now());
            return "product.jpg";
        });
        CompletableFuture<String> futureAttr = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询商品的属性"+ LocalTime.now());
            return "黑色8+128G";
        });
        CompletableFuture<String> futureDesc = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询商品介绍"+ LocalTime.now());
            return "最强续航222天信号绕全球五圈";
        });
        CompletableFuture<Void> allOf = CompletableFuture.allOf(futureAttr, futureDesc, futureImg);
        allOf.get();
        System.out.println("查询完毕"+ LocalTime.now());
    }
}